#include <sys/types.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>

#include "wqueue.h"
#include "request.h"
#include "response.h"
#include "log.h"

#define DEFAULT_WQUEUE_PORT 6017
#define DEFAULT_MAX_THREADS 16

static queue_thread *th_pool;

static int wqueue_exit = 0;
static int servfd;

static int thread_number;

pthread_mutex_t wexit = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t wlock = PTHREAD_MUTEX_INITIALIZER;

int  make_wpool();

void main_wqueue(void* i);

void lock_wqueue_main()
{
again:
	if(0 != pthread_mutex_lock(&wexit))
	{
		if(errno == EINTR) goto again;
		fprintf(stderr, "call pthread_mutex_lock wexit failed!\n");
	}
}

static int network_listen(int port)
{
	int fd, flag = 1;
	struct sockaddr_in addr;
	struct linger ling;

	if((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
	{
		perror("socket");
		return -1;
	}

	memset(&addr, 0x0, sizeof(struct sockaddr_in));
	addr.sin_family = AF_INET;
	addr.sin_port   = htons(port);
	addr.sin_addr.s_addr = htons(INADDR_ANY);	
	if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(int)) == -1)
	{
		perror("setsockopt");
		return -1;		
	}

	memset(&ling, 0x0, sizeof(ling));
	ling.l_onoff  = 1;
	ling.l_linger = 30;
	if(setsockopt(fd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)) == -1)
	{
		perror("setsockopt");
		return -1;		
	}

	if((flag = fcntl(fd, F_SETFL, 0)) == -1
		|| fcntl(fd, F_SETFL, flag|O_NONBLOCK) == -1)
	{
		perror("fcntl");
		return -1;
	}

	if(bind(fd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) == -1)
	{
		perror("bind");
		return -1;
	}

	if(listen(fd, thread_number + 10) == -1)
	{
		perror("listen");
		return -1;
	}
	
	return fd;
}

int do_connect_process(tcp_client *client)
{
	connection *connect = connection_init(NULL, client);
	connect->response.http_status = do_request_process(connect);
	connect->response.version = connect->request.version;
	if(status_drops_connection(connect->response.http_status)) {
		client->ready = client->keep_alive = 0;
		do_response_process(connect);
		connection_free(&connect);
		return 1;
	}

	QueueErr ret;
	queue_header header;
	queue_header_init(&header);
	header.size = connect->request.content_length;
	logger(PRIORITY_ACCESS, "%s\t%s", inet_ntoa(client->addr.sin_addr), connect->request.path);
	if(Q_EXEC_OK != (ret = wqueue_append(&header, connect->request.content, connect->request.path))) {
		if(ret == QERR_CMD_UNKNOWN) {
			connect->response.http_status = HTTP_NOT_FOUND;
		}
		else
		{
			logger(PRIORITY_ERROR, "wqueue append record Failed %s\t%s\t%d",
				inet_ntoa(client->addr.sin_addr), connect->request.path);
			connect->response.http_status = HTTP_INTERNAL_SERVER_ERROR; 
		}
	}

	if(do_response_process(connect) == -1) {
		client->ready = client->keep_alive = 0;
		connection_free(&connect);
		return -1;
	}
	
	connection_free(&connect);
	client->ready = 1;
	return 1;
}

void wqueue_start()
{
	int port, i, count;
	const char *pport, *ppth;
	pthread_t th;

	wqueue_exit = 0;
	if(0 != pthread_create(&th, NULL, (void*)lock_wqueue_main, NULL)) {
		fprintf(stderr, "call lock_wqueue failure!\n");
		return;
	}
	pthread_join(th, NULL);

	pport = INI_GetValue(cfg, "global", "QueueWritePort");
	ppth  = INI_GetValue(cfg, "global", "MaxQWriteThread");
	port  = pport ? atoi(pport) : DEFAULT_WQUEUE_PORT;
	thread_number = ppth  ? atoi(ppth)  : DEFAULT_MAX_THREADS;
	
	if((servfd = network_listen(port)) < 0)
		exit(1);

	if(make_wpool() < 0)
		exit(1);
again:
	if(0 != pthread_mutex_lock(&wexit))
	{
		if(errno == EINTR) goto again;
		logger(PRIORITY_ERROR, "call pthread_mutex_lock failure at %d int '%s'", __LINE__, __FILE__);
	}
	
	shutdown(servfd, SHUT_RDWR);
	for(i = 0; i < thread_number; i++)
	{
#ifdef DEBUG
		fprintf(stderr, "waiting for wqueue index: %d", i);
#endif
		count = 0;
		while(th_pool[i].runing == 1)
		{
#ifdef DEBUG
			fprintf(stderr, ".");
#endif
			pthread_mutex_unlock(&wlock);
			usleep(50000);
			if(count++ == 60) {
#ifdef DEBUG
				fprintf(stderr, " not exit in 3s.");
#endif
				break;
			}
		}
#ifdef DEBUG
		fprintf(stderr, "\n");
#endif
		pthread_join(th_pool[i].tid, NULL);
	}

	free(th_pool);
#ifdef DEBUG
	fprintf(stderr, "wqueue end!\n");
#endif
}

void wqueue_stop()
{
	wqueue_exit = 1;
	pthread_mutex_unlock(&wexit);
}

int make_wpool()
{
	int i;
	th_pool = (queue_thread*)malloc(thread_number*sizeof(queue_thread));
	if(th_pool == NULL)
	{
		fprintf(stderr, "call malloc th_pool error!\n");
		return -1;
	}

	for(i = 0; i < thread_number; i++)
	{
		if(0 != pthread_create(&th_pool[i].tid, NULL, (void*)main_wqueue, (void*)i)) {
			fprintf(stderr, "call pthread_create failed!\n");
			free(th_pool);
			return -2;
		}
	}

	return 1;
}

void main_wqueue(void* i)
{
	int flag, index = (int)i;
	tcp_client client;

	fd_set rfds, srfds, rexp, sexp;
	struct timeval tv;
	tv.tv_sec  = 30;
	tv.tv_usec = 0;
	th_pool[index].runing = 1;

start_main:
	FD_ZERO(&rfds);
	srfds = rfds;
	if(0 != pthread_mutex_lock(&wlock))
	{
		if(errno == EINTR) goto start_main;
		logger(PRIORITY_ERROR, "call pthread_mutex_lock failed at %d int '%s'", __LINE__, __FILE__);
		goto terminate;
	}
	if(wqueue_exit) goto terminate;

#ifdef DEBUG
	fprintf(stderr, "wqueue thread[%d] on select.\n", index);
#endif
	FD_SET(servfd, &srfds);
	sexp = srfds;
	if(!select(servfd+1, &srfds, NULL, &sexp, NULL)
		|| !FD_ISSET(servfd, &srfds) || FD_ISSET(servfd, &sexp))
	{
		logger(PRIORITY_DEBUG, "%s at %d int '%s'", strerror(errno), __LINE__, __FILE__);
		goto start_main;
	}
#ifdef DEBUG
	fprintf(stderr, "wqueue thread[%d] on accept.\n", index);
#endif
	memset(&client, 0x0, sizeof(client));
	client.addr_len = sizeof(client.addr);
	client.sock = accept(servfd, (struct sockaddr*)&(client.addr), &(client.addr_len));

	while(0 != pthread_mutex_unlock(&wlock))
	{
		if(errno == EINTR) continue;
		logger(PRIORITY_ERROR, "call pthread_mutex_unlock failed at %d int '%s'", __LINE__, __FILE__);
		goto terminate;
	}

	if(-1 == client.sock)
	{
		if(errno == EINTR) goto start_main;
		logger(PRIORITY_ERROR, "call accept failed at %d int '%s'", __LINE__, __FILE__);
		goto terminate;
	}

	if((flag = fcntl(client.sock, F_SETFL, 0)) == -1
		|| fcntl(client.sock, F_SETFL, flag|O_NONBLOCK) == -1)
	{
		logger(PRIORITY_ERROR, "%s at %d int '%s'", strerror(errno), __LINE__, __FILE__);
		goto terminate;
	}

	FD_SET(client.sock, &rfds);
	rexp = rfds;
	if(!select(client.sock+1, &rfds, NULL, &rexp, &tv)
		|| !FD_ISSET(client.sock, &rfds) || FD_ISSET(client.sock, &rexp))
	{
		close(client.sock);
		goto start_main;
	}

	th_pool[index].count++;
	if(-1 == do_connect_process(&client))
		logger(PRIORITY_ERROR, "do request failed at %d in '%s'", __LINE__, __FILE__);

	/* TODO handle keep_alive. */
	if(-1 == close(client.sock))
		logger(PRIORITY_ERROR, "close client %d failed at %d in '%s'", client.sock, __LINE__, __FILE__);

	goto start_main;

terminate:
	th_pool[index].runing = 0;
}
